package com.sunxd.zookeeper.use.lock;

import com.alibaba.fastjson.JSON;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * @author: 作者名称
 * @date: 2021-07-05 15:24
 **/
@Slf4j
@Data
public class ZkLockWatcher implements AsyncCallback.ChildrenCallback, AsyncCallback.StringCallback, Watcher, AsyncCallback.StatCallback {

    private static final String ZK_PREFIX = "/";

    private CountDownLatch countDownLatch;

    private ZooKeeper zookeeper;

    private String path;

    private String date;

    private String ephemeralPath;

    private String threadName;
    private String unionkey;


    /**
     * 尝试锁
     */
    public void tryLock() {
        try {
            byte[] bytes = zookeeper.getData("/", Boolean.FALSE, new Stat());
            if(unionkey.equals(new String(bytes))){
                log.info("线程：{} 当前锁正在执行，无需重新获得锁--",threadName);
            }else {
                log.info(" {} create node ....",threadName );
                countDownLatch = new CountDownLatch(1);
                zookeeper.create(path, date.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, "");
                countDownLatch.await();
            }

        } catch (Exception e) {
            log.info("---zk create ephemeral error ,path:{} , bytes:{} ,e:{} ", path, date, e);
        }
    }

    public void unlock() throws KeeperException, InterruptedException {
        log.info("线程：{} 释放锁--",threadName);
        zookeeper.delete(ephemeralPath, -1);
    }


    /**
     * 创建临时节点回掉
     *
     * @param rc
     * @param path 路径
     * @param ctx
     * @param name 临时节点的名字
     */
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if (StringUtils.isNotBlank(name)) {
            this.ephemeralPath = name;
            log.info("--：{} create node name:{}",threadName, name);
            zookeeper.getChildren( "/", Boolean.FALSE, this, ctx);
        }
    }

    /**
     * 获取子节点回掉
     *
     * @param rc
     * @param path
     * @param ctx
     * @param list
     */
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> list) {
//        log.info("--获取子节点信息，eg:第一个countdown，执行业务，最后释放锁，eg:后面的则监听前一个节点的事件--");
        // 乱序-需要排序
        list.sort(Comparator.naturalOrder());
//        log.info("---getChildren2 list  rc:{}, path:{} , ctx:{} , list:{}", rc, path, ctx, JSON.toJSONString(list));
        if (!CollectionUtils.isEmpty(list)) {
            int indexOf = list.indexOf(ephemeralPath.substring(1));
            if (indexOf == 0) {
                log.info("-----线程：{} 我是第一个，获得了锁",threadName);
                try {
                    zookeeper.setData("/",unionkey.getBytes(),-1);
                } catch (Exception e) {
                    log.info("将当前信息放入父目录失败。。。");
                }
                countDownLatch.countDown();
            } else {
                log.info("-----线程：{} 我不是第一个，我在监听上一个节点-----",threadName);
                zookeeper.exists(ZK_PREFIX + list.get(indexOf - 1), this, this, ctx);
            }
        }
    }


    @Override
    public void process(WatchedEvent watchedEvent) {
        switch (watchedEvent.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                zookeeper.getChildren( "/", Boolean.FALSE, this, "ctx");
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
            case PersistentWatchRemoved:
                break;
        }
    }

    /**
     * exists state call back
     * @param i
     * @param s
     * @param o
     * @param stat
     */
    @Override
    public void processResult(int i, String s, Object o, Stat stat) {

    }
}
